#include <iostream>
#include <bits/stdc++.h>
#include <sys/time.h>
#include <time.h>
#include <unistd.h>
#include <chrono>
#include <unistd.h>
#include <fcntl.h>
#include "../buttonrpc/buttonrpc.hpp"
#include <mutex>
#include <thread>
#include <condition_variable>
#include <atomic>
#include <algorithm>

using namespace std;
#define COMMOM_PORT 1234
#define HEART_BEART_PERIOD 100

// 阻塞队列
template <typename T>
class MessageQueue {
public:
    explicit MessageQueue(int capacity = 1) : capacity_(capacity) {}

    void Push(const T& msg) {
        std::unique_lock<std::mutex> lock(mutex_);
        condvar_.wait(lock, [this]() { return queue_.size() < capacity_;});
        queue_.push(msg);
        condvar_.notify_all();
    }

    pair<T, bool> TPop() {
        std::unique_lock<std::mutex> lock(mutex_);
        auto start_time = std::chrono::steady_clock::now();
        while (queue_.empty()) {
            if (condvar_.wait_for(lock, std::chrono::milliseconds(300)) == std::cv_status::timeout) {
                // 等待超时，跳出循环并返回空
                return make_pair(T{}, false); // 默认构造一个空结构
            }
        }

        T msg = queue_.front();
        queue_.pop();
        condvar_.notify_all();
        return make_pair(msg, ture);
    }

    T Pop() {
        std::unique_lock<std::mutex> lock(mutex_);
        condvar_.wait(lock, [this](){ return !queue_.empty(); });

        T msg = queue_.front();
        queue_.pop();
        condvar_.notify_all();
        return make_pair(msg, ture);
    }


    bool empty(){
        std::unique_lock<std::mutex> lock(mutex_);
        return queue_.empty();
    }

    void setcapacity(int capacity){
        std::unique_lock<std::mutex> lock(mutex_);
        capacity_ = capacity;
    }

private:
    std::queue<T> queue_;
    std::mutex mutex_;
    std::condition_variable condvar_;
    int capacity_;
};

//通过传入raft.start()得到的返回值
struct StartState{
    int startIndex;
    int startTerm;
    bool isLeader;

    StartState():startIndex(-1), startTerm(-1), isLeader(false){}
};

struct Operation{
    string getCmd();
    string op;
    string key;
    string value;
    int clientId;
    int requestId;
    int term;
    int index;

    string getCmd(){
        string cmd = op + " " + key + " " + value + " " + to_string(clientId) + " " + to_string(requestId);
        return cmd;
    }
};

struct ApplyMsg{
    bool commandValid;
	string command;
    int commandIndex;
    int commandTerm;
    Operation getOperation();

    ApplyMsg::ApplyMsg(){
        commandTerm = -1;
        commandIndex = -1;
    }

    Operation getOperation(){
        Operation operation;
        vector<string> str;
        string tmp;
        for(int i = 0; i < command.size(); i++){
            if(command[i] != ' '){
                tmp += command[i];
            }else{
                if(tmp.size() != 0) str.push_back(tmp);
                tmp = "";
            }
        }
        if(tmp.size() != 0){
            str.push_back(tmp);
        }
        operation.op = str[0];
        operation.key = str[1];
        operation.value = str[2];
        operation.clientId = atoi(str[3].c_str());
        operation.requestId = atoi(str[4].c_str());
        operation.term = commandTerm;
        return operation;
    }
};


//一个存放当前raft的ID及自己两个RPC端口号的结构(为了减轻负担，一个选举，一个日志同步，分开来)
struct PeersInfo{
    pair<int, int> m_port;
    int m_peerId;
};

//日志结构
struct LogEntry{
    string command;  // 命令
    int    term;     // 任期

    LogEntry(string cmd = "", int term = -1):command(cmd),term(term){}
};

// 持久化结构——在响应RPC之前，在稳定的存储上更新
struct Persister{
    int              currentTerm;  // 服务器看到的最新任期
    int              votedFor;     // 当前任期内获得选票的候选人ID
    vector<LogEntry> log;          // 日志条目；每个条目包含状态机的命令，以及领导者收到条目的时间
};

enum RAFT_STATE {LEADER = 0, CANDIDATE, FOLLOWER};          // 用枚举定义的raft三种状态

// -------------------------------RPC参数-------------------------------------
// RequestVote RPC 候选人为收集选票而调用
// Arguments: 
struct RequestVoteArgs{
    int term;          // 候选人的任期号
    int candidateId;   // 请求选票的候选人的 Id
    int lastLogIndex;  // 候选人的最后日志条目的索引值
    int lastLogTerm;   // 候选人最后日志条目的任期号
};
// Results: 
struct RequestVoteReply{
    int  term;         // 当前任期号，候选人会更新自己的任期号
    bool VoteGranted;  // true 表示候选人获得了选票
};

// AppendEntries RPC 由领导者调用以复制日志条目；也作为心跳使用
// Arguments: 
struct AppendEntriesArgs{
    int    term;          // 领导任期
    int    leaderId;      // 使追随者可以为客户端重定向
    int    prevLogIndex;  // 紧接在新日志之前的日志条目的索引
    int    prevLogTerm;   // 紧邻新日志条目之前的日志条目的任期
    int    leaderCommit;  // 领导者的已知已提交的最高的日志条目的索引
    string sendLog;       // 需要被保存的日志条目（做心跳使用时，内容为空；为了提高效率可一次性发送多个）

    friend Serializer& operator >> (Serializer& in, AppendEntriesArgs& d) {
		in >> d.term >> d.leaderId >> d.prevLogIndex >> d.prevLogTerm >> d.leaderCommit >> d.sendLog;
		return in;
	}
	friend Serializer& operator << (Serializer& out, AppendEntriesArgs d) {
		out << d.term << d.leaderId << d.prevLogIndex << d.prevLogTerm << d.leaderCommit << d.sendLog;
		return out;
	}
};
// Results: 
struct AppendEntriesReply{
    int  term;            // 当前任期，领导者会更新自己的任期
    bool success;         // 如果跟随者所含有的条目和prevLogIndex以及prevLogTerm匹配，则为真
    int  conflict_term;   // 用于冲突时日志快速匹配
    int  conflict_index;  // 用于冲突时日志快速匹配                 
};


class Raft{
public:
    Raft() = default;
    static void* listenForVote(void* arg);              // 用于监听voteRPC的server线程
    static void* listenForAppend(void* arg);            // 用于监听appendRPC的server线程
    static void* appendEntriesTicker(void* arg);        // 持续处理日志同步的守护线程
    static void* electionTicker(void* arg);             // 持续处理选举的守护线程
    static void* callRequestVote(void* arg, int id);    // 发voteRPC的线程
    static void* sendAppendEntries(void* arg, int id);  // 发appendRPC的线程
    static void* applyLogLoop(void* arg);               // 与上层KVServer进行交互的线程

    void Make(vector<PeersInfo> peers, int id);                 // raft初始化
    int getduration(timeval last);                              // 传入某个特定计算到当下的持续时间
    void setBroadcastTime();                                    // 重新设定BroadcastTime，成为leader发心跳的时候需要重置
    pair<int, bool> getState();                                 // 判断是否leader
    RequestVoteReply requestVote(RequestVoteArgs args);         // vote的RPChandler
    AppendEntriesReply appendEntries(AppendEntriesArgs args);   // append的RPChandler
    bool checkLogUptodate(int term, int index);                 // 判断是否最新日志(两个准则)，vote时会用到
    vector<LogEntry> getCmdAndTerm(string text);                // 用的RPC不支持传容器，所以封装成string，这个是解封装恢复函数
    StartState start(string command);                           // 向raft传日志的函数，只有leader响应并立即返回，应用层用到
    int getlastIndex();
    int getlastTerm();

    void PushMQ(ApplyMsg msg){
        MQ.Push(msg);
    }
    pair<ApplyMsg, bool> TPopMQ(){
        return MQ.TPop();
    }
    ApplyMsg PopMQ(){
        return MQ.Pop();
    }
    
    void serialize();      // 序列化
    bool deserialize();    // 反序列化
    void saveRaftState();  // 持久化
    void readRaftState();  // 读取持久化状态
    bool isKilled();       // ->check is killed?
    void kill();           // 设定raft状态为dead
    void Raft::activate(); // 重新激活raft
    ~Raft() = default;
    
private:
    mutex r_lock;                 // 锁
    condition_variable cond;      // 条件变量
    vector<PeersInfo> peers;      // 投票信息
    Persister persister;          // 持久化信息
    int peerId;                   // 当前节点id
    atomic_bool dead;

    // 需要持久化的数据
    int currentTerm; 
    int votedFor;
    vector<LogEntry> log;

    // 所有服务器上的易失性状态
    int commitIndex;  // 已知被提交的最高日志条目的索引（初始化为0，单调增加）
    int lastApplied;  // 已应用于状态机的最高日志条目的索引（初始化为0，单调增加）

    // 领导者（服务器）上的易失性状态
    vector<int> nextIndex;   // 对于每个服务器，要发送给该服务器的下一个日志条目的索引（初始化为领导者的最后一个日志索引+1）
    vector<int> matchIndex;  // 对于每个服务器，已知在服务器上复制的最高日志条目的索引（初始化为0，单调增加）

    int recvVotes;     // 收到的票数
    int finishedVote;  // 已投票的节点数量

    RAFT_STATE state;
    int leaderId;
    struct timeval lastWakeTime;
    struct timeval lastBroadcastTime;

    MessageQueue<ApplyMsg> MQ;  // 在applyLogLoop中存msg，每次存一条，处理完再存一条
};